package Autodesk;


import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Hashtable;
import java.util.List;
import java.util.Map;
import java.util.StringTokenizer;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.filecache.DistributedCache;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
import org.apache.log4j.Logger;

public class ParseLogs extends Configured implements Tool  {
		private static Logger logger = Logger.getLogger(ParseLogs.class);
		public Path prodPath;
		public static class LogMapper extends Mapper<Object, Text, Text, IntWritable> {
			
			private static final String EXCLUDE_VALUES = "beehive_monitor";
			private static final String EXCLUDE_VALUES2 = "your monitor";  
			private final static IntWritable one = new IntWritable(1);
			private Text word = new Text();
			private Map<ArrayList<String>,String> StaticListLang=new HashMap<ArrayList<String>,String>();
			private Map<ArrayList<String>,String> StaticListProd=new HashMap<ArrayList<String>,String>();
			private Map<ArrayList<String>,String> StaticListChannel=new HashMap<ArrayList<String>,String>();
			private Map<ArrayList<String>,String> StaticListCountry=new HashMap<ArrayList<String>,String>();
			private Map<ArrayList<String>,String> StaticListIp=new HashMap<ArrayList<String>,String>();
			private ArrayList<String> listChannelsExcluded=new ArrayList<String>();
//			private Map<ArrayList<String>,String> StaticListFeature=new HashMap<ArrayList<String>,String>();
			Parser par=new Parser();
			

			public void setup(Context context){
				
				Configuration conf=context.getConfiguration();
				 if (conf.getBoolean("ip.patterns",false )){
					  Path[] patternsFiles = new Path[0];
					  try{
				           patternsFiles = DistributedCache.getLocalCacheFiles(conf);
					  }catch(IOException ioe){
				           System.err.println("Caught exception while getting cached files: ");
					  }
					  for (Path patternsFile : patternsFiles) {
						  par.getStaticList(patternsFile,StaticListIp,"ip"); 
			 				logger.info("Loaded Ips-------------------");
					  }
				 }
		 				StaticListLang=par.getStaticList("language"); 
		 				logger.info("Loaded Langs-------------------");
				  		
				  		StaticListProd=par.getStaticList("product"); 
				  		logger.info("Loaded Products-------------------");
				  		
				  		StaticListChannel=par.getStaticList("channel"); 
						System.out.println(StaticListChannel);

				  		logger.info("Loaded Channels-------------------");
				  		
				  		StaticListCountry=par.getStaticList("country");
		 				logger.info("Loaded Countries-------------------");

		 				//				  		StaticListFeatures=par.getStaticList("features");
//				  		System.out.println("Loaded Features-------------------");
				 inicitializeListChannelsExc();

			}


			private void inicitializeListChannelsExc() {
				listChannelsExcluded.add("productchannel");
				 listChannelsExcluded.add("productchannel2");
				 listChannelsExcluded.add("supportchannel");
				 listChannelsExcluded.add("supportchannel2");
				 listChannelsExcluded.add("acadconnectchannel");
				 listChannelsExcluded.add("acadconnectchannel2");
				 listChannelsExcluded.add("taxonomy3");

			}
						
		private String getParameter(String parameter,Hashtable<String, String> ht){
			String result;
			if ((ht.get(parameter) != null) && !(ht.get(parameter).isEmpty()))
				result=ht.get(parameter).toString();
			else
				result="2";
			return result;
		}

		private String getQueryResult(String line){
			String result="0";
			if(line.indexOf("x-query-result=")>-1){
				line = line.substring(line.indexOf("x-query-result=") + 15);
				result = line.substring(0,line.indexOf(" "));
				if(result.equals("")){
					result="0";
				}
			}
			return result;
		}
		

		private StringBuffer getOutput(StringBuffer bf,Hashtable<String, String> ht,String date,String channel,String ipPort,
				String hour,String source,String queryResult,String country){

			bf.append(date + "\t");//Date
			bf.append(channel.equals(null) ? "2" : channel);//Channel
			bf.append("\t");
			bf.append(ipPort + "\t");//IpPort
			bf.append(getParameter("p",ht));//Products
			bf.append("\t");
			bf.append(getParameter("l",ht));//Languages
			bf.append("\t");
			bf.append(hour);//Hour
			bf.append("\t");
			bf.append(getParameter("q",ht));
			bf.append("\t");
			bf.append(queryResult);//QueryResult
			bf.append("\t");
			bf.append(country);//Country
//			bf.append("\t");
//			bf.append(getParameter("s",ht));
//			bf.append("\t");
//			bf.append(getParameter("aud",ht));
//			bf.append("\t");
//			bf.append(getParameter("cg",ht));
//			bf.append("\t");
//			bf.append(getParameter("origin",ht));
//			bf.append("\t");
//			bf.append(getParameter("sc",ht));
//			bf.append("\t");
//			bf.append(getParameter("c",ht));
			
			return bf;
		}

		
		//MAP FUNCTION
		public void map(Object key, Text value, Context context) throws IOException, InterruptedException {

			String line = value.toString().toLowerCase();
			Hashtable<String, String> ht = new Hashtable<String, String>();
			String date = null;
			String channel = null;
			String source=null;
			ArrayList<String> listIpForw=new ArrayList<String>();
			// Looking for DATE
			date = par.getDate(line);
			String hour=par.getHour(line);
			
			// Prepare Line
			line = line.toLowerCase();
			
			// Looking for IP.
			String ipPort = par.getIp(line);
			if(ipPort.isEmpty()){
				ipPort="2";
			}
			
			// Looking for listForward.
			listIpForw=par.getListIPForwaredFor(line);
			String ipExcluded="1";
			if(!listIpForw.isEmpty()){
				for(int h=0;h<listIpForw.size();h++){
					ipExcluded=par.lookForKey(StaticListIp, listIpForw.get(h), "ip");
				}
			}
			
			// Looking for CHANNEL.
			String channelID;	
			channel = par.getChannel(line);
			channelID= par.lookForKey(StaticListChannel, channel,"channel");
			 //Looking for Country
			String country=par.getCountry(line);
			country=par.lookForKey(StaticListCountry,country,"country");

			//LOOKING FOR QueryResult
			String queryResult=getQueryResult(line);//Amount of times that the specified query appears 
//			if(line.indexOf("acadconnectchannel3")>-1 && (line.indexOf(EXCLUDE_VALUES)==-1)){
//				System.out.println("1_"+channel);
//				System.out.println("2-"+channelID);
//			//	System.out.println("3 IP-"+ipExcluded);
//				System.out.println("3 beehive-"+line.indexOf(EXCLUDE_VALUES));
//				System.out.println(line);
//			}

			 if( !(date.equals("0")) && (!channelID.equals("1")) && (ipExcluded.equals("1")) 
					 && (line.indexOf(EXCLUDE_VALUES)==-1) && (line.indexOf(EXCLUDE_VALUES2)==-1)){
//			if( !(date.equals("0")) && (!channelID.equals("1")) && (line.indexOf(EXCLUDE_VALUES)==-1)){
//				if(line.indexOf("acadconnectchannel3")>-1){
//					System.out.println("4_"+channel);
//				}
				String queryLine = par.getQueryLine(line);
					StringTokenizer tokenizer = new StringTokenizer(queryLine,"&");
					while (tokenizer.hasMoreTokens()) {
						String token = tokenizer.nextToken();
							
							if (token.indexOf("q=") > -1) {
								ht.put("q", token.substring(2));
							} else if (token.indexOf("p=") > -1) {
								try{
									String prodAb=par.getProduct(token, queryLine,par);								
									String[] productRele=prodAb.split("\t");
									String tokenProd=productRele[0];
									String tokenRele=productRele[1];
									String id=par.lookForKeyProduct(StaticListProd, tokenProd.toLowerCase(),tokenRele.toLowerCase());
									ht.put("p", id);
								}catch(Exception e){
									ht.put("p", "2");
								}
							} else if ((token.indexOf("l=") > -1) && (token.indexOf("channel=") < 0) && (token.indexOf("url=") < 0)) {

								String tokenSub = token.substring(2);								
								String id = par.lookForKey(StaticListLang, tokenSub,"language");
								ht.put("l", id);
								
//							} else if (token.indexOf("source=") > -1) {
//								token = token.substring(token.indexOf("=")+1);
//								String id = par.lookForKey(StaticListFeature, token,"features");
//								ht.put("s", id);
//							}else if (token.indexOf("audience=") > -1) {
//								token = token.substring(token.indexOf("=")+1);
//								String id = par.lookForKey(StaticListFeature, token,"features");
//								ht.put("aud", id);
//							}else if (token.indexOf("sc=") > -1) {
//								token = token.substring(token.indexOf("=")+1);
//								String id = par.lookForKey(StaticListFeature, token,"features");
//								ht.put("sc",id);	
//
//							}else if (token.indexOf("c=") > -1) {
//								token = token.substring(token.indexOf("=")+1);
//								String id = par.lookForKey(StaticListFeature, token,"features");
//								ht.put("c", id);
//
//							}else if (token.indexOf("cg=") > -1) {
//								token = token.substring(token.indexOf("=")+1);
//								String id = par.lookForKey(StaticListFeature, token,"features");
//								ht.put("cg",id);
//							}else if (token.indexOf("origin=") > -1) {
//								token = token.substring(token.indexOf("=")+1);
//								String id = par.lookForKey(StaticListFeature, token,"features");
//								ht.put("origin", id);
							}
						}
				}
				StringBuffer bf = new StringBuffer();
				if (ht != null && !ht.isEmpty() ) {
//					if(line.indexOf("acadconnectchannel3")>-1){
//						System.out.println("5_"+channel);
//					}
					bf = getOutput(bf, ht, date, channelID, ipPort,hour,source,queryResult,country);
					word.set(bf.toString());
					context.write(word, one);

				}
		}

	}

	//REDUCE FUNCTION	
	public static class LogReducer extends
			Reducer<Text, IntWritable, Text, IntWritable> {
		
		private IntWritable result = new IntWritable();
//		private MultipleOutputs<Text,IntWritable> mos;
//		String pathName; 
		
//		 public void setup(Context context) {
//			 Configuration conf=context.getConfiguration();
//			 pathName=conf.get("pathName");
//			 mos = new MultipleOutputs<Text,IntWritable>(context);
			 
//		 }
		public void reduce(Text key, Iterable<IntWritable> values,
				Context context) throws IOException, InterruptedException {
			
			int sum = 0;
			for (IntWritable val : values) {
				sum += val.get();
			}
			result.set(sum);
			context.write(key, result);
			//mos.write(key, result, pathName);
		
		}
//		 public void cleanup(Context context) throws IOException, InterruptedException {
//			 mos.close();
//		 }
	}
	
	public int run(String[] args) throws Exception { 
		Configuration conf = getConf();
		
		Job job = new Job(conf, "ParseLogs");
		job.setJarByClass(Autodesk.ParseLogs.class);
		
		
		job.setMapperClass(LogMapper.class);
		job.setReducerClass(LogReducer.class);
		
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(IntWritable.class);
		
		List<String> other_args = new ArrayList<String>();
		for (int i = 0; i < args.length; ++i) {
			if ("-ip".equals(args[i])) {
				prodPath=new Path(args[++i]);
				DistributedCache.addCacheFile(prodPath.toUri(),
						job.getConfiguration());
				job.getConfiguration().setBoolean("ip.patterns",
						true);
				job.getConfiguration().set("ip.path", prodPath.getName());
			}else{
				other_args.add(args[i]);
			}
		}

	    FileInputFormat.setInputPaths(job, new Path(other_args.get(0)));
	    FileOutputFormat.setOutputPath(job, new Path(other_args.get(1)));
	    
//	    String inputPathName=other_args.get(0).toString();
//	    inputPathName=inputPathName.substring(inputPathName.indexOf("_S.")+3);
//	    System.out.println(inputPathName);
//	    job.getConfiguration().set("pathName", inputPathName);
	    
//	    MultipleOutputs.addNamedOutput(job,"ParsedLogsReport", TextOutputFormat.class, Text.class, IntWritable.class);
		System.exit(job.waitForCompletion(true)?0:1);
		
		return 0;
	
	}
	
	public static void main(String[] args) throws Exception {
		int res = ToolRunner.run(new Configuration(), new ParseLogs(), args);
		System.exit(res);
		}

 
}